package kafka_learn

import (
	"fmt"
	"github.com/IBM/sarama"
)

func ProductSync() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.ClientID = "client01"
	config.Net.MaxOpenRequests = 10

	broker := []string{"127.0.0.1:9092", "127.0.0.1:9101"}

	producter, err := sarama.NewSyncProducer(broker, config)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := producter.Close(); err != nil {
			panic(err)
		}
	}()
	for i := 0; i < 100; i++ {
		part, offset, err := producter.SendMessage(&sarama.ProducerMessage{
			Topic: "test01",
			Value: sarama.StringEncoder(fmt.Sprintf("hello-world-%d", i)),
		})
		fmt.Println(i, part, offset, err)
	}

}

func ProductAsync() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.ClientID = "client01"
	config.Net.MaxOpenRequests = 10

	broker := []string{"127.0.0.1:9092", "127.0.0.1:9101"}

	producter, err := sarama.NewAsyncProducer(broker, config)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := producter.Close(); err != nil {
			panic(err)
		}
	}()

	go func() {
		for {
			select {
			case msg := <-producter.Successes():
				fmt.Println("success", msg.Key, msg.Value, msg.Offset)
			case err = <-producter.Errors():
				fmt.Println("err", err)
			}
		}
	}()

	for i := 0; i < 10; i++ {
		producter.Input() <- &sarama.ProducerMessage{
			Topic: "test01",
			Value: sarama.StringEncoder(fmt.Sprintf("hello-b-%d", i)),
		}
	}

	over := make(chan bool)
	<-over
}
